package beautiful.butterfly.drds.data_exchange.taskgroup.runner;

import beautiful.butterfly.drds.data_exchange.data_exchange.exchanger.Produce;
import beautiful.butterfly.drds.data_exchange.plugin.AbstractTaskPlugin;
import beautiful.butterfly.drds.data_exchange.read_and_write_data.AbstractReader;
import beautiful.butterfly.drds.data_exchange.report.Messages;
import beautiful.butterfly.drds.data_exchange.statistics.Phase;
import beautiful.butterfly.drds.data_exchange.statistics.RunningInfo;
import lombok.extern.slf4j.Slf4j;


@Slf4j
public class ReadeTask extends Task implements Runnable
{
    private Produce produce;

    public ReadeTask(AbstractTaskPlugin abstractTaskPlugin)
    {
        super(abstractTaskPlugin);
    }

    public void setProduce(Produce produce)
    {
        this.produce = produce;
    }

    public void run()
    {
        assert null != this.produce;

        AbstractReader.AbstractTask abstractTask = (AbstractReader.AbstractTask) this.getTaskPlugin();


        RunningInfo runningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.wait_write_time);
        try
        {
            runningInfo.start();

            log.debug("task reade starts to do init ...");
            RunningInfo initRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.read_task_init);
            initRunningInfo.start();
            abstractTask.init();
            initRunningInfo.end();

            log.debug("task reade starts to do prepare ...");
            RunningInfo prepareRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.read_task_prepare);
            prepareRunningInfo.start();
            abstractTask.prepare();
            prepareRunningInfo.end();

            log.debug("task reade starts to read ...");
            RunningInfo dataRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.read_task_data);
            dataRunningInfo.start();
            abstractTask.startRead(produce);
            produce.terminate();

            dataRunningInfo.addCount(Messages.getTotalReadRecords(super.getMessage()));
            dataRunningInfo.addSize(Messages.getTotalReadBytes(super.getMessage()));
            dataRunningInfo.end();

            log.debug("task reade starts to do post ...");
            RunningInfo postRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.read_task_post);
            postRunningInfo.start();
            abstractTask.post();
            postRunningInfo.end();
            // automatic pushAllToRecordQueue
            // super.markSuccess(); 这里不能标记为成功，成功的标志由 writerRunner 来标志（否则可能导致 reade 先结束，而 write 还没有结束的严重 bug）
        } catch (Throwable e)
        {
            log.error("AbstractReader runner Received Exceptions:", e);
            super.markFail(e);
        } finally
        {
            log.debug("task reade starts to do destroy ...");
            RunningInfo destroyRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.read_task_destroy);
            destroyRunningInfo.start();
            super.destroy();
            destroyRunningInfo.end();

            runningInfo.end(super.getMessage().getLong(Messages.wait_writer_time));

            long transformerUsedTime = super.getMessage().getLong(Messages.transformer_used_time);
            if (transformerUsedTime > 0)
            {
                RunningInfo transformerRecord = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.transformer_time);
                transformerRecord.start();
                transformerRecord.end(transformerUsedTime);
            }
        }
    }

    public void shutdown()
    {
        produce.shutdown();
    }
}
